package utils;


import domain.EventCatagoryProductCount;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Create by zhh on 2022/9/16.
 * * @param <IN> The type of the input value.
 * * @param <OUT> The type of the output value.
 * * @param <KEY> The type of the key.
 * * @param <W> The type of {@code Window} that this window function can be applied on.
 */
public class TopNWindowFunction implements WindowFunction<Long, EventCatagoryProductCount, Tuple3<String, String, String>, TimeWindow> {
    @Override
    public void apply(Tuple3<String, String, String> value, TimeWindow window, Iterable<Long> input, Collector<EventCatagoryProductCount> out) throws Exception {
        String event = value.f0;
        String category = value.f1;
        String product= value.f2;
        Long count = input.iterator().next();
        //FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss");//设置时间打印的格式
//        String start = format.format(window.getStart());
//        String end = format.format(window.getEnd());
        long start = window.getStart();
        long end = window.getEnd();
        out.collect(new EventCatagoryProductCount(event,category,product,count,start,end));
    }
}
